package com.mytijian.task.task;

import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import com.mytijian.base.exception.BizException;
import com.mytijian.base.exception.ErrorInfo;
import com.mytijian.task.enums.TaskStatusEnum;
import com.mytijian.task.httprequest.HttpRequest;
import com.mytijian.task.model.Result;
import com.mytijian.task.model.SubTask;
import com.mytijian.task.model.SubmitTask;
import com.mytijian.task.task.step.TaskStep;
import com.mytijian.task.task.step.TaskStepContainer;
import com.mytijian.task.zookeeper.LeaderService;
import com.mytijian.task.enums.SubTaskStatusEnum;
import com.mytijian.task.enums.TaskTypeEnum;
import com.mytijian.task.model.Task;
import com.mytijian.task.repository.RuningTaskRepository;
import com.mytijian.task.repository.TaskRepository;
import com.mytijian.task.task.handout.DefaultHandoutController;
import com.mytijian.task.task.handout.HandoutController;
import com.mytijian.task.zookeeper.ZookeeperRegistryCenter;
import com.mytijian.task.zookeeper.path.NodePath;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.web.ServerProperties;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Service;

import java.util.*;

@Service("taskOperateService")
public class TaskOperateServiceImpl implements TaskOperateService {
    private static String workNode = "/workNode/";
    private Logger logger = LoggerFactory.getLogger(TaskOperateServiceImpl.class);
    @Autowired
    private TaskRepository taskRepository;
    @Autowired
    private List<AbstractTaskProcess> taskProcesses;
    @Autowired
    private LeaderService leaderService;
    @Autowired
    private RuningTaskRepository runingTaskRepository;
    @Autowired
    private ServerProperties serverProperties;
    private HandoutController getHandoutController() {
        return new DefaultHandoutController(zookeeperRegistryCenter,leaderService , runingTaskRepository, taskRepository);
    }
    @Autowired
    private ZookeeperRegistryCenter zookeeperRegistryCenter;

    @Override
    public Result submit(SubmitTask submitTask) {
        ErrorInfo errorInfo = TaskStepContainer.beforeSubmitValidate(submitTask);
        if (errorInfo != null) {
            throw new BizException(errorInfo);
        }
        Object response = TaskStepContainer.reader(submitTask.getTaskTypeEnum(), submitTask.getContent());
        if (response == null) {
            throw new BizException("12", "读取的数据不能为null");
        }

        Task task = new Task();
        BeanUtils.copyProperties(submitTask, task);
        task.setStatus(TaskStatusEnum.un_begin.getCode());
        if (response instanceof Collection) {
            Collection collection = (Collection) response;
            task.setTotalNum(collection.size());
            List<SubTask> subTasks = new ArrayList<>(collection.size());
            int taskId = taskRepository.insert(task);
            for (Object o : collection) {
                SubTask subTask = buildSubTask(taskId, o);
                subTasks.add(subTask);
            }
            taskRepository.batchInsert(subTasks);
            getHandoutController().handout(taskId,submitTask, subTasks);
            return null;
        }
        if (response instanceof Map) {
            Map map = (Map) response;
            task.setTotalNum(map.size());
            int taskId = taskRepository.insert(task);
            List<SubTask> subTasks = new ArrayList<>(map.size());
            map.forEach((key, value) -> {
                Map<Object, Object> temp = new HashMap<>();
                temp.put(key, value);
                buildSubTask(taskId, JSONObject.toJSONString(temp));
            });
            taskRepository.batchInsert(subTasks);
            getHandoutController().handout(taskId,submitTask, subTasks);
            return null;
        }
        task.setTotalNum(1);
        List<SubTask> subTasks = new ArrayList<>();
        int taskId = taskRepository.insert(task);
        subTasks.add(buildSubTask(taskId, response));
        taskRepository.batchInsert(subTasks);
        getHandoutController().handout(taskId,submitTask, subTasks);
        return null;
    }

    @Override
    public void stop(Integer taskId) {
        validateEndStatus(taskId);
        taskRepository.updateTaskStatus(taskId, TaskStatusEnum.stop.getCode());
        Map<String, ResponseEntity> responses =operate("/stop", taskId);
        validateResponse(responses);
    }

    @Override
    public void suspend(Integer taskId) {
        validateEndStatus(taskId);
        taskRepository.updateTaskStatus(taskId, TaskStatusEnum.suspend.getCode());

        Map<String, ResponseEntity> jsonObjects = operate("/suspend", taskId);
        validateResponse(jsonObjects );
    }

    @Override
    public void process(Integer taskId) {
        validateEndStatus(taskId);
        taskRepository.updateTaskStatus(taskId, TaskStatusEnum.processing.getCode());

        Map<String, ResponseEntity> jsonObjects = operate("/process", taskId);
        validateResponse(jsonObjects);
    }

    private SubTask buildSubTask(Integer taskId, Object o) {
        SubTask subTask = new SubTask();
        subTask.setTaskId(taskId);
        subTask.setStatus(SubTaskStatusEnum.un_begin.getCode());
        subTask.setParam(JSONObject.toJSONString(o));
        subTask.setParamClass(o.getClass().getName());
        return subTask;
    }

    private void validateEndStatus(Integer taskId) {
        Task task = taskRepository.getTasksByIds(Lists.newArrayList(taskId)).get(0);
        if (TaskStatusEnum.getEndStatusInteger().contains(task.getStatus())) {
            throw new BizException("12", "任务已经处于终态:" + TaskStatusEnum.getByCode(task.getStatus()).getDesc()+"不能执行");
        }
    }
    private void validateResponse(Map<String, ResponseEntity> responseEntityMap) {
        StringBuilder builder = new StringBuilder();
        responseEntityMap.forEach((ipAndPort, jsonObject)->{
            if (responseEntityMap.get(ipAndPort).getStatusCodeValue() != 200) {
                builder.append(ipAndPort).
                        append("节点操作失败:").
                        append(responseEntityMap.get(ipAndPort).getBody()).append('\n');
            }
        });
        if (builder.length() > 0) {
            throw new BizException("12",builder.toString());
        }
    }
    private Map<String, ResponseEntity> operate(String operate, Integer taskId) {
        List<String> ipAndPorts =zookeeperRegistryCenter.getChildrenKeys(NodePath.node);
        Map<String, ResponseEntity> response = new HashMap<>();
        for (String ipAndPort : ipAndPorts) {
            // 这里不存在资源竞争，所以不需要主节点
            // 分配给工作节点，工作节点自己检查是否有这个任务，如果有这个任务，执行相应的操作
            Map<String, Object> param = getParams(taskId);
            ResponseEntity responseEntity = null;
            responseEntity = getRequest(ipAndPort, operate, param);

            response.put(ipAndPort, responseEntity);
        }
        return response;
    }
    private Map<String, Object> getParams(Integer taskId) {
        Map<String, Object> param = new HashMap<>();
        param.put("taskId", taskId);
        return param;
    }
    private ResponseEntity postRequest(String ipAndPort, String operate, Object body) {
        return new HttpRequest(serverProperties.getServlet().getContextPath()).postFoRequest(ipAndPort,workNode + operate, body);
    }
    private ResponseEntity getRequest(String ipAndPort, String operate, Map<String, Object> param) {
        return new HttpRequest(serverProperties.getServlet().getContextPath()).getForRequest(ipAndPort,workNode +operate, param);

    }
    private AbstractTaskProcess getTaskProcess(TaskTypeEnum taskTypeEnum) {
//        for (AbstractTaskProcess taskProcess : taskProcesses) {
//            if (taskProcess.accept(taskTypeEnum)) {
//                return taskProcess;
//            }
//        }
        throw new BizException("123", "不存在");
    }

}
